﻿using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Ddl;
using Me.Prettyprint.Hector.Api.Factory;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Connection
{
    /// <summary>
    /// NodeDiscovery has the actual logic of figuring out when a host
    /// is new to the ring.
    /// </summary>
    public class NodeDiscovery
    {
        private Logger logger = new Logger(typeof(NodeDiscovery));
        private CassandraHostConfigurator cassandraHostConfigurator;
        private HConnectionManager connectionManager;
        private DataCenterValidator dataCenterValidator;

        public NodeDiscovery(CassandraHostConfigurator cassandraHostConfigurator, HConnectionManager connectionManager)
        {
            this.cassandraHostConfigurator = cassandraHostConfigurator;
            this.connectionManager = connectionManager;
            this.dataCenterValidator = new DataCenterValidator(cassandraHostConfigurator.AutoDiscoveryDataCenters);
        }
        public void AddNodes()
        {
            List<CassandraHost> foundhosts = DiscoverNodes();
            if (foundhosts != null && foundhosts.Count > 0)
            {
                foreach (CassandraHost cassandraHost in foundhosts)
                {
                    cassandraHostConfigurator.ApplyConfig(cassandraHost);
                    connectionManager.AddCassandraHost(cassandraHost);
                }
            }
        }
        public List<CassandraHost> DiscoverNodes()
        {
            Cluster cluster = HFactory.GetOrCreateCluster(connectionManager.ClusterName,this.cassandraHostConfigurator);
            if (cluster == null)
            {
                return null;
            }
            List<CassandraHost> existinghosts = connectionManager.GetHosts();
            List<CassandraHost> found = new List<CassandraHost>();
            try {
                foreach (KeyspaceDefinition keyspaceDefinition in cluster.DescribeKeyspaces())
                {
                    if (!keyspaceDefinition.Name.Equals(KeyspaceFields.KEYSPACE_SYSTEM))
                    {
                        List<TokenRange> tokenRanges = cluster.DescribeRing(keyspaceDefinition.Name);
                        foreach (TokenRange tokenRange in tokenRanges)
                        {
                            foreach (EndpointDetails endPointDetail in tokenRange.Endpoint_details)
                            {
                                if (dataCenterValidator == null || dataCenterValidator.Validate(endPointDetail.Datacenter))
                                {
                                    CassandraHost foundHost = new CassandraHost(endPointDetail.Host, cassandraHostConfigurator.Port);
                                    if (!existinghosts.Contains(foundHost))
                                    {
                                        logger.Info(string.Format("Found a node we don't know about {0} for TokenRange {1}", foundHost.Ip, tokenRange)); 
                                        found.Add(foundHost);
                                    }
                                }
                            }
                        }
                        break;
                    }
                }
            }
            catch (Exception e)
            {
                logger.Error(e.StackTrace);
            }
            return found;
        }
    }
}
